#include <assert.h>
#include <errno.h>
#include <netinet/in.h>
#include <search.h>
#include <signal.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <sys/eventfd.h>

#include "shm_ipc.h"
#ifdef _XOPEN_SOURCE_EXTENDED
#include <arpa/inet.h>
#endif
#include <pthread.h>
#include <sys/socket.h>

#include <map>

#include "deamon.h"
#include "game_event.h"
#include "kafka_thread.h"
#include "logic_thread.h"
#include "oper_config.h"
#include "time_helper.h"
#include "msg_queue.h"
#include "librdkafka/rdkafka.h"
//#include "../proto/db.pb-c.h"

#define TIMEOUT_MISEC 10

char *     g_conn_topic;
int        g_conn_partid;
static int game_kafka_recv_callback(struct rd_kafka_message_s *rkm)
{
	send_to_logic(rkm, rkm->len);
	wakeup_logic_thread(0);	
	return (0);
}
static int game_logic_callback(void *data)
{
	struct rd_kafka_message_s *rkm = (struct rd_kafka_message_s *)data;
	send_msg((char *)"echo back: ", 11, g_conn_topic, g_conn_partid);	
	send_msg((char *)rkm->payload, rkm->len, g_conn_topic, g_conn_partid);
	wakeup_send_thread(0);
	return (0);
}

int main(int argc, char **argv)
{
	int   ret  = 0;
	FILE *file = NULL;
	char *line;
	char *conn_brokers;
	int   game_partid;
	char *game_topic;
	char *game_brokers;
	pthread_t kafka_recv_pid;
	pthread_t kafka_send_pid;	

	signal(SIGTERM, SIG_IGN);

	ret = log4c_init();
	if (ret != 0)
	{
		printf("log4c_init failed[%d]\n", ret);
		return (ret);
	}

	init_mycat();
	if (!mycat)
	{
		printf("log4c_category_get(\"six13log.log.app.application1\"); failed\n");
		return (0);
	}
	for (int i = 1; i < argc; ++i)
	{
		if (strcmp(argv[i], "-d") == 0)
		{
			change_to_deamon();
			//			break;
		}
		else if (strcmp(argv[i], "-t") == 0)
		{  /// test for mem check
			open_err_log_file();
		}
	}

	uint64_t pid = write_pid_file();
	LOG_INFO("%s %d: game_srv run %lu", __FUNCTION__, __LINE__, pid);

	ret = game_event_init();
	if (ret != 0)
		goto done;

	file = fopen("../server_info.ini", "r");
	if (!file)
	{
		LOG_ERR("open server_info.ini failed[%d]", errno);
		ret = -1;
		goto done;
	}

	line = get_first_key(file, (char *)"game_topic");
	game_topic = get_value(line);
	line = get_first_key(file, (char *)"game_partid");
	game_partid = atoi(get_value(line));
	line = get_first_key(file, (char *)"game_broker");
	game_brokers = get_value(line);
	line = get_first_key(file, (char *)"conn_topic");
	g_conn_topic = get_value(line);
	line = get_first_key(file, (char *)"conn_partid");
	g_conn_partid = atoi(get_value(line));
	line = get_first_key(file, (char *)"conn_broker");
	conn_brokers = get_value(line);

	if (SIG_ERR == signal(SIGPIPE, SIG_IGN))
	{
		LOG_ERR("set sigpipe ign failed");
		return (0);
	}

	fclose(file);
	file = NULL;

	kafka_recv_pid = create_kafka_recv_thread(game_brokers, (char *)"1", game_partid, game_topic, game_kafka_recv_callback, NULL);
	kafka_send_pid = create_kafka_send_thread(conn_brokers, NULL);

	init_thread_logic_param(game_logic_callback);
	thread_logic(NULL);

	pthread_join(kafka_recv_pid, NULL);
	pthread_join(kafka_send_pid, NULL);	

	LOG_INFO("srv loop stoped[%d]", ret);
	
done:
	if (file)
		fclose(file);
	if (g_conn_topic)
		free(g_conn_topic);
	LOG_INFO("game_srv stoped[%d]", ret);
	return (ret);
}
